package org.example.rabbitmq.client.routing;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class DirectRoute {
    private static final String EXCHANGE_NAME = "direct_route";
    ExecutorService executor = Executors.newFixedThreadPool(3);
    List<String> routingKeys = new ArrayList<>();

    public void shutdown() {
        executor.shutdown();
    }

    public void publish(int times) throws IOException, TimeoutException {
        waitAMinute();
        Connection connection = null;
        try {
            connection = newConnection();
            Channel channel = connection.createChannel();
            declareFanoutExchange(channel);
            int count = routingKeys.size();
            for (int i = 0; i < times; i++) {
                String routingKey = routingKeys.get(i % count);
                String message = routingKey + " message " + i;
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println("[x] Sent " + message);
            }
        } finally {
            closeConnection(connection);
        }
    }

    public void registerListener(String routingKey) {
        routingKeys.add(routingKey);
        executor.execute(() -> {
            CountDownLatch latch = new CountDownLatch(1);
            Connection connection = null;
            try {
                connection = newConnection();
                Channel channel = connection.createChannel();
                String queueName = declareQueueAndBind(channel, routingKey);
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String message = new String(body, "UTF-8");
                        System.out.println("[x] " + routingKey + " consumer received messages: " + message);
//                channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                };
                // 当autoAck为true时，不再需要channel.basicAck
                channel.basicConsume(queueName, true, consumer);
                System.out.println("[*] Waiting for messages.");
                latch.await(5, TimeUnit.SECONDS);
            } catch (IOException | TimeoutException | InterruptedException ex) {
                System.out.println("[x] Error: " + ex.getMessage());
            } finally {
                closeConnection(connection);
            }
        });
    }

    private void waitAMinute() {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private Connection newConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest@local");
        return factory.newConnection();
    }

    private String declareQueueAndBind(Channel channel, String routingKey) throws IOException {
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, true, null);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
        return queueName;
    }

    private void declareFanoutExchange(Channel channel) throws IOException {
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, true, null);
    }

    private void closeConnection(Connection connection) {
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
